package cn.doitedu.rtdw.datasync;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class JOB01_注册信息表同步HBASE {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://doitedu:8020/rtdw_job1_ckpt");
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射mysql的源表
        tenv.executeSql(
                " CREATE TABLE ums_member_mysql (      "+
                        "   id bigint                          "+
                        "   ,member_level_id bigint            "+
                        "   ,username string                   "+
                        "   ,password string                   "+
                        "   ,nickname string                   "+
                        "   ,phone string                      "+
                        "   ,status int                        "+
                        "   ,create_time timestamp(3)          "+
                        "   ,icon string                       "+
                        "   ,gender int                        "+
                        "   ,birthday date                     "+
                        "   ,city string                       "+
                        "   ,job string                        "+
                        "   ,personalized_signature string     "+
                        "   ,source_type int                   "+
                        "   ,integration int                   "+
                        "   ,growth int                        "+
                        "   ,luckey_count int                  "+
                        "   ,history_integration int           "+
                        "   ,modify_time timestamp(3)          "+
                        "   ,primary key(id)  not enforced     "+
                        " ) with (                             "+
                        "      'connector' = 'mysql-cdc',      "+
                        "      'hostname' = 'doitedu',         "+
                        "      'port' = '3306',                "+
                        "      'username' = 'root',            "+
                        "      'password' = 'root',            "+
                        "      'database-name' = 'realtimedw', "+
                        "      'table-name' = 'ums_member'     "+
                        " )                                    "
        );


        // 建表，映射hbase的目标表
        tenv.executeSql(
                " create table user_hbase(                 "+
                        " 	user_name string,                      "+
                        "   f  row<                                " +
                        "          id  bigint                      "+
                        " 	       ,member_level_id bigint         "+
                        "          ,password string                "+
                        "          ,nickname string                "+
                        "          ,phone string                   "+
                        "          ,status int                     "+
                        "          ,create_time timestamp(3)       "+
                        "          ,icon string                    "+
                        "          ,gender int                     "+
                        "          ,birthday date                  "+
                        "          ,city string                    "+
                        "          ,job string                     "+
                        "          ,personalized_signature string  "+
                        "          ,source_type int                "+
                        "          ,integration int                "+
                        "          ,growth int                     "+
                        "          ,luckey_count int               "+
                        "          ,history_integration int        "+
                        "          ,modify_time timestamp(3)       "+
                        " 	>                                      "+
                        " ) WITH (                                 "+
                        "  'connector' = 'hbase-2.2',              "+
                        "  'table-name' = 'doit42_user_info',      "+
                        "  'zookeeper.quorum' = 'doitedu:2181'     "+
                        " )                                        "
        );


        // 写一个sql
        tenv.executeSql(
                " INSERT INTO user_hbase                    "+
                        " SELECT                                      "+
                        "   username,                                 "+
                        "   row(id,member_level_id,                   "+
                        "   password,nickname,                        "+
                        "   phone,status,                             "+
                        "   create_time,                              "+
                        "   icon,gender,birthday,                     "+
                        "   city,job,personalized_signature,          "+
                        "   source_type,integration,                  "+
                        "   growth,luckey_count,                      "+
                        "   history_integration,modify_time)  as f    "+
                        " FROM ums_member_mysql                "
        );
    }

}
